package com.atguigu.flink.timeAndwindow;

import com.atguigu.flink.pojo.Event;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;

/**
 * Created by 黄凯 on 2023/6/16 0016 18:03
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 */
public class Flink01_UserDefineWaterMark {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.getConfig().setAutoWatermarkInterval(1000);

        SingleOutputStreamOperator<Event> ds = env.socketTextStream("127.0.0.1", 8888)
                .map(

                        line -> {

                            String[] fields = line.split(",");

                            System.out.println("fields = " + Arrays.toString(fields));

                            return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));

                        }

                );

        ds.print("input");

        //生成水位线
        ds.assignTimestampsAndWatermarks(

                new WatermarkStrategy<Event>() {


                    @Override
                    public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {

                        return new WatermarkGenerator<Event>() {

                            //2s的延迟
                            private Long delay = 2000L;

                            //已到数据的最大时间戳
                            private Long maxTs = Long.MIN_VALUE + delay;

                            /**
                             * 每条数据都生成水位线
                             * @param event
                             * @param eventTimestamp
                             * @param output
                             */
                            @Override
                            public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {

                                //有序流每条数据都生成水位线
//                                output.emitWatermark(new Watermark(event.getTs()));
//                                maxTs = event.getTs();

                                //乱序流每条数据生成水位线

//                                System.out.println("eventTimestamp = " + eventTimestamp);
//
//                                maxTs = Math.max(maxTs, eventTimestamp);
//                                System.out.println("发射的水位线：" + maxTs);
//                                output.emitWatermark(new Watermark(maxTs));

                                //乱序流每条数据生成水位线 + 迟到数据
//                                maxTs = Math.max(maxTs, eventTimestamp);
//                                System.out.println("发射的水位线：" + (maxTs - delay));
//                                output.emitWatermark(new Watermark(maxTs - delay));

                                maxTs = Math.max(maxTs, eventTimestamp);

                            }

                            /**
                             * 周期性发射水位线，默认是200ms
                             * PipelineOptions.AUTO_WATERMARK_INTERVAL
                             *
                             * 设置周期：
                             *   env.getConfig().setAutoWatermarkInterval(1000);
                             *
                             * @param output
                             */
                            @Override
                            public void onPeriodicEmit(WatermarkOutput output) {

                                //有序流周期性生成水位线
//                                output.emitWatermark(new Watermark(maxTs));
//                                System.out.println("发射的水位线：" + maxTs);

                                //乱序流周期性生成水位线
//                                System.out.println("发射的水位线：" + maxTs);
//                                output.emitWatermark(new Watermark(maxTs));

                                //乱序流周期性生成水位线 + 迟到数据
                                System.out.println("发射的水位线：" + (maxTs - delay));
                                output.emitWatermark(new Watermark(maxTs - delay));

                            }
                        };


                    }

                    /**
                     * 创建时间戳提取器对象，TimestampAssigner
                     * extractTimestamp() 从数据中提取时间戳
                     * @param context
                     * @return
                     */
                    @Override
                    public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {

                        return new TimestampAssigner<Event>() {


                            /**
                             * 从数据中提取时间戳
                             * @param element The element that the timestamp will be assigned to.
                             * @param recordTimestamp The current internal timestamp of the element, or a negative value, if
                             *     no timestamp has been assigned yet.
                             * @return
                             */
                            @Override
                            public long extractTimestamp(Event element, long recordTimestamp) {

                                return element.getTs();


                            }
                        };

                    }
                }

        );

        env.execute();

    }

}
